#include "asycipc/named_pipe.hpp"

#ifdef _WIN32
#ifndef NOMINMAX
#define NOMINMAX
#endif
#include <windows.h>

#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstring>
#include <deque>
#include <limits>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <utility>
#include <vector>

#include "asycipc/api.hpp"
#include "asycipc/frame.hpp"

namespace asycipc {

class WindowsPipeHandle : public PipeHandle {
 public:
  WindowsPipeHandle(HANDLE handle, bool overlapped = false);
  ~WindowsPipeHandle() override;

  void Write(const std::vector<uint8_t>& data) override;
  std::vector<uint8_t> Read() override;

  HANDLE native_handle() const noexcept { return handle_; }
  HANDLE Detach() noexcept;
  bool overlapped() const noexcept { return overlapped_; }

 private:
  HANDLE handle_ = INVALID_HANDLE_VALUE;
  bool overlapped_ = false;
};

class WindowsPipeListener : public PipeListener {
 public:
  explicit WindowsPipeListener(std::wstring pipe_name,
                               std::size_t buffer_size = 4096);
  ~WindowsPipeListener() override = default;

  PipeHandlePtr Accept() override;
  const std::wstring& pipe_name() const noexcept { return pipe_name_; }

 private:
  std::wstring pipe_name_;
  std::size_t buffer_size_;
};

namespace {

std::string NarrowFromWide(const std::wstring& value) {
  if (value.empty()) {
    return {};
  }
  const int length =
      WideCharToMultiByte(CP_UTF8, 0, value.c_str(), -1, nullptr, 0, nullptr, nullptr);
  if (length <= 0) {
    return {};
  }
  std::string result(static_cast<std::size_t>(length - 1), '\0');
  WideCharToMultiByte(CP_UTF8, 0, value.c_str(), -1, result.data(), length, nullptr,
                      nullptr);
  return result;
}

std::wstring WideFromUtf8(const std::string& value) {
  if (value.empty()) {
    return {};
  }
  const int length =
      MultiByteToWideChar(CP_UTF8, 0, value.c_str(), -1, nullptr, 0);
  if (length <= 0) {
    return {};
  }
  std::wstring result(static_cast<std::size_t>(length - 1), L'\0');
  MultiByteToWideChar(CP_UTF8, 0, value.c_str(), -1, result.data(), length);
  return result;
}

void ThrowLastError(const std::wstring& where) {
  const DWORD error = GetLastError();
  LPWSTR buffer = nullptr;
  FormatMessageW(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM |
                     FORMAT_MESSAGE_IGNORE_INSERTS,
                 nullptr, error, MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
                 reinterpret_cast<LPWSTR>(&buffer), 0, nullptr);
  std::wstring wmessage = buffer ? buffer : L"unknown";
  if (buffer) {
    LocalFree(buffer);
  }
  const std::string location = NarrowFromWide(where);
  const std::string message = NarrowFromWide(wmessage);
  throw NamedPipeError(location + " failed: " + message);
}

enum class OperationType { kRead, kWrite };

struct ConnectionContext;

struct SharedIocpTelemetry {
  std::atomic<int> active_connections{0};
};

struct IocpOperation {
  OVERLAPPED overlapped{};
  OperationType type = OperationType::kRead;
  std::shared_ptr<ConnectionContext> context;
  std::vector<uint8_t> buffer;

  IocpOperation(OperationType op_type, std::shared_ptr<ConnectionContext> ctx)
      : type(op_type), context(std::move(ctx)) {}
};

struct ConnectionContext : std::enable_shared_from_this<ConnectionContext> {
  PipeServer& server;
  std::unique_ptr<WindowsPipeHandle> handle;
  std::unique_ptr<Channel> channel;
  HANDLE completion_port = nullptr;
  std::chrono::milliseconds maintenance_interval{1000};
  std::size_t read_buffer_size = 4096;
  std::thread maintenance_thread;
  std::mutex write_mutex;
  std::deque<std::vector<uint8_t>> write_queue;
  bool write_in_progress = false;
  std::atomic<bool> closing{false};
  std::atomic<int> active_operations{0};
  std::mutex maintenance_mutex;
  bool maintenance_joined = false;
  TelemetryCenter* telemetry = nullptr;
  std::shared_ptr<SharedIocpTelemetry> shared;
  std::atomic<bool> closure_reported{false};

  ConnectionContext(PipeServer& srv, std::unique_ptr<WindowsPipeHandle> pipe_handle,
                    std::unique_ptr<Channel> channel_ptr, HANDLE port,
                    std::chrono::milliseconds maintenance, std::size_t read_buffer,
                    std::shared_ptr<SharedIocpTelemetry> telemetry_shared)
      : server(srv),
        handle(std::move(pipe_handle)),
        channel(std::move(channel_ptr)),
        completion_port(port),
        maintenance_interval(maintenance),
        read_buffer_size(read_buffer),
        telemetry(&srv.telemetry()),
        shared(std::move(telemetry_shared)) {}

  void StartMaintenance();
  void StopMaintenance();
  void ReportConnectionOpen();
  void ReportConnectionClosed();
  void UpdatePendingOpsGauge();
  void SetWriteQueueGauge(std::size_t size);
  void RecordReadBytes(std::size_t bytes);
  void RecordWriteBytes(std::size_t bytes);
};

void QueueFrames(const std::shared_ptr<ConnectionContext>& ctx,
                 const std::vector<Frame>& frames);
void CloseConnection(const std::shared_ptr<ConnectionContext>& ctx,
                     const std::string& reason);

void ConnectionContext::StartMaintenance() {
  auto weak = weak_from_this();
  maintenance_thread = std::thread([weak]() {
    while (auto ctx = weak.lock()) {
      if (ctx->closing.load(std::memory_order_relaxed)) {
        break;
      }
      std::this_thread::sleep_for(ctx->maintenance_interval);
      if (ctx->closing.load(std::memory_order_relaxed)) {
        break;
      }
      auto frames = ctx->server.CollectMaintenanceFrames(*ctx->channel);
      if (!frames.empty()) {
        QueueFrames(ctx, frames);
      }
    }
  });
}

void ConnectionContext::StopMaintenance() {
  std::lock_guard<std::mutex> lock(maintenance_mutex);
  if (maintenance_thread.joinable() && !maintenance_joined) {
    maintenance_thread.join();
    maintenance_joined = true;
  }
}

void ConnectionContext::ReportConnectionOpen() {
  if (!telemetry) {
    return;
  }
  telemetry->IncrementCounter("ipc_winpipe_connection_open_total");
  int active = 1;
  if (shared) {
    active = shared->active_connections.fetch_add(1, std::memory_order_relaxed) + 1;
    active = std::max(active, 0);
  }
  telemetry->SetGauge("ipc_winpipe_connections_active",
                      static_cast<double>(active));
}

void ConnectionContext::ReportConnectionClosed() {
  if (closure_reported.exchange(true)) {
    return;
  }
  if (telemetry) {
    telemetry->IncrementCounter("ipc_winpipe_disconnect_total");
  }
  int active = 0;
  if (shared) {
    active = shared->active_connections.fetch_sub(1, std::memory_order_relaxed) - 1;
    active = std::max(active, 0);
  }
  if (telemetry) {
    telemetry->SetGauge("ipc_winpipe_connections_active",
                        static_cast<double>(active));
  }
}

void ConnectionContext::UpdatePendingOpsGauge() {
  if (!telemetry) {
    return;
  }
  const int pending = std::max(active_operations.load(std::memory_order_relaxed), 0);
  telemetry->SetGauge("ipc_winpipe_pending_ops", static_cast<double>(pending));
}

void ConnectionContext::SetWriteQueueGauge(std::size_t size) {
  if (!telemetry) {
    return;
  }
  telemetry->SetGauge("ipc_winpipe_write_queue_depth",
                      static_cast<double>(size));
}

void ConnectionContext::RecordReadBytes(std::size_t bytes) {
  if (!telemetry || bytes == 0) {
    return;
  }
  telemetry->IncrementCounter("ipc_winpipe_read_bytes_total",
                              static_cast<uint64_t>(bytes));
}

void ConnectionContext::RecordWriteBytes(std::size_t bytes) {
  if (!telemetry || bytes == 0) {
    return;
  }
  telemetry->IncrementCounter("ipc_winpipe_write_bytes_total",
                              static_cast<uint64_t>(bytes));
}

void ReportIocpError(const std::shared_ptr<ConnectionContext>& ctx,
                     const std::string& reason) {
  if (ctx->telemetry) {
    ctx->telemetry->IncrementCounter("ipc_winpipe_io_error_total");
  }
  ctx->server.ReportError(reason);
}

void FinalizeIfIdle(const std::shared_ptr<ConnectionContext>& ctx);

bool SubmitRead(const std::shared_ptr<ConnectionContext>& ctx) {
  if (ctx->closing.load(std::memory_order_relaxed)) {
    return false;
  }
  auto* op = new IocpOperation(OperationType::kRead, ctx);
  op->buffer.resize(ctx->read_buffer_size);
  std::memset(&op->overlapped, 0, sizeof(op->overlapped));
  ctx->active_operations.fetch_add(1, std::memory_order_relaxed);
  ctx->UpdatePendingOpsGauge();
  const BOOL ok = ReadFile(ctx->handle->native_handle(), op->buffer.data(),
                           static_cast<DWORD>(op->buffer.size()), nullptr,
                           &op->overlapped);
  if (!ok) {
    const DWORD error = GetLastError();
    if (error == ERROR_BROKEN_PIPE || error == ERROR_PIPE_NOT_CONNECTED) {
      ctx->active_operations.fetch_sub(1, std::memory_order_relaxed);
      ctx->UpdatePendingOpsGauge();
      delete op;
      CloseConnection(ctx, "pipe disconnected");
      return false;
    }
    if (error != ERROR_IO_PENDING) {
      ctx->active_operations.fetch_sub(1, std::memory_order_relaxed);
      ctx->UpdatePendingOpsGauge();
      delete op;
      ReportIocpError(ctx, "ReadFile failed");
      CloseConnection(ctx, "ReadFile failed");
      return false;
    }
  }
  return true;
}

bool SubmitNextWriteLocked(const std::shared_ptr<ConnectionContext>& ctx) {
  if (ctx->write_in_progress || ctx->write_queue.empty() ||
      ctx->closing.load(std::memory_order_relaxed)) {
    return true;
  }
  auto payload = std::move(ctx->write_queue.front());
  ctx->write_queue.pop_front();
  ctx->SetWriteQueueGauge(ctx->write_queue.size());
  ctx->write_in_progress = true;

  auto* op = new IocpOperation(OperationType::kWrite, ctx);
  op->buffer = std::move(payload);
  std::memset(&op->overlapped, 0, sizeof(op->overlapped));
  ctx->active_operations.fetch_add(1, std::memory_order_relaxed);
  ctx->UpdatePendingOpsGauge();
  const BOOL ok = WriteFile(ctx->handle->native_handle(), op->buffer.data(),
                            static_cast<DWORD>(op->buffer.size()), nullptr,
                            &op->overlapped);
  if (!ok) {
    const DWORD error = GetLastError();
    if (error != ERROR_IO_PENDING) {
      ctx->active_operations.fetch_sub(1, std::memory_order_relaxed);
      ctx->UpdatePendingOpsGauge();
      ctx->write_in_progress = false;
      delete op;
      ReportIocpError(ctx, "WriteFile failed");
      return false;
    }
  }
  return true;
}

void QueueFrames(const std::shared_ptr<ConnectionContext>& ctx,
                 const std::vector<Frame>& frames) {
  if (frames.empty() || ctx->closing.load(std::memory_order_relaxed)) {
    return;
  }
  bool submit_failed = false;
  {
    std::lock_guard<std::mutex> lock(ctx->write_mutex);
    for (const auto& frame : frames) {
      ctx->write_queue.push_back(frame.Serialize());
    }
    ctx->SetWriteQueueGauge(ctx->write_queue.size());
    if (!ctx->write_in_progress) {
      if (!SubmitNextWriteLocked(ctx)) {
        submit_failed = true;
      }
    }
  }
  if (submit_failed) {
    CloseConnection(ctx, "write submission failed");
  }
}

void CloseConnection(const std::shared_ptr<ConnectionContext>& ctx,
                     const std::string& reason) {
  bool expected = false;
  if (!ctx->closing.compare_exchange_strong(expected, true)) {
    return;
  }
  ctx->ReportConnectionClosed();
  CancelIoEx(ctx->handle->native_handle(), nullptr);
  {
    std::lock_guard<std::mutex> lock(ctx->write_mutex);
    ctx->write_queue.clear();
    ctx->SetWriteQueueGauge(0);
    ctx->write_in_progress = false;
  }
  if (reason != "pipe disconnected" && reason != "peer closed" &&
      reason != "shutdown") {
    ReportIocpError(ctx, reason);
  }
  FinalizeIfIdle(ctx);
}

void FinalizeIfIdle(const std::shared_ptr<ConnectionContext>& ctx) {
  if (!ctx->closing.load(std::memory_order_relaxed)) {
    return;
  }
  if (ctx->active_operations.load(std::memory_order_relaxed) == 0) {
    ctx->StopMaintenance();
    ctx->SetWriteQueueGauge(0);
    ctx->UpdatePendingOpsGauge();
    ctx->ReportConnectionClosed();
    if (ctx->handle) {
      ctx->handle.reset();
    }
  }
}

void HandleReadCompletion(IocpOperation* op, DWORD transferred) {
  auto ctx = op->context;
  ctx->active_operations.fetch_sub(1, std::memory_order_relaxed);
  ctx->UpdatePendingOpsGauge();
  if (ctx->closing.load(std::memory_order_relaxed)) {
    delete op;
    FinalizeIfIdle(ctx);
    return;
  }
  if (transferred == 0) {
    delete op;
    CloseConnection(ctx, "peer closed");
    FinalizeIfIdle(ctx);
    return;
  }
  op->buffer.resize(transferred);
  ctx->RecordReadBytes(transferred);
  auto frames = ctx->server.ProcessInbound(*ctx->channel, op->buffer);
  delete op;
  if (!frames.empty()) {
    QueueFrames(ctx, frames);
  }
  if (!ctx->closing.load(std::memory_order_relaxed)) {
    SubmitRead(ctx);
  }
  FinalizeIfIdle(ctx);
}

void HandleWriteCompletion(IocpOperation* op) {
  auto ctx = op->context;
  const std::size_t bytes_written = op->buffer.size();
  ctx->active_operations.fetch_sub(1, std::memory_order_relaxed);
  ctx->UpdatePendingOpsGauge();
  ctx->RecordWriteBytes(bytes_written);
  delete op;
  if (ctx->closing.load(std::memory_order_relaxed)) {
    FinalizeIfIdle(ctx);
    return;
  }
  bool submit_failed = false;
  {
    std::lock_guard<std::mutex> lock(ctx->write_mutex);
    ctx->write_in_progress = false;
    if (!ctx->write_queue.empty()) {
      if (!SubmitNextWriteLocked(ctx)) {
        submit_failed = true;
      }
    }
    ctx->SetWriteQueueGauge(ctx->write_queue.size());
  }
  if (submit_failed) {
    CloseConnection(ctx, "write submission failed");
  }
  FinalizeIfIdle(ctx);
}

}  // namespace

WindowsPipeHandle::WindowsPipeHandle(HANDLE handle, bool overlapped)
    : handle_(handle), overlapped_(overlapped) {
  if (handle_ == INVALID_HANDLE_VALUE) {
    throw NamedPipeError("invalid pipe handle");
  }
}

WindowsPipeHandle::~WindowsPipeHandle() {
  if (handle_ != INVALID_HANDLE_VALUE) {
    CloseHandle(handle_);
    handle_ = INVALID_HANDLE_VALUE;
  }
}

HANDLE WindowsPipeHandle::Detach() noexcept {
  HANDLE handle = handle_;
  handle_ = INVALID_HANDLE_VALUE;
  return handle;
}

void WindowsPipeHandle::Write(const std::vector<uint8_t>& data) {
  if (handle_ == INVALID_HANDLE_VALUE) {
    throw NamedPipeError("pipe closed");
  }
  if (!overlapped_) {
    DWORD written = 0;
    const uint8_t* cursor = data.data();
    DWORD remaining = static_cast<DWORD>(data.size());
    while (remaining > 0) {
      if (!WriteFile(handle_, cursor, remaining, &written, nullptr)) {
        ThrowLastError(L"WriteFile");
      }
      remaining -= written;
      cursor += written;
    }
    return;
  }

  HANDLE event_handle = CreateEventW(nullptr, TRUE, FALSE, nullptr);
  if (!event_handle) {
    ThrowLastError(L"CreateEventW");
  }
  OVERLAPPED overlapped{};
  overlapped.hEvent = event_handle;
  DWORD written = 0;
  BOOL ok = WriteFile(handle_, data.data(), static_cast<DWORD>(data.size()),
                      nullptr, &overlapped);
  if (!ok) {
    const DWORD error = GetLastError();
    if (error != ERROR_IO_PENDING) {
      CloseHandle(event_handle);
      ThrowLastError(L"WriteFile");
    }
    if (WaitForSingleObject(event_handle, INFINITE) != WAIT_OBJECT_0) {
      CloseHandle(event_handle);
      ThrowLastError(L"WaitForSingleObject");
    }
  }
  if (!GetOverlappedResult(handle_, &overlapped, &written, TRUE)) {
    CloseHandle(event_handle);
    ThrowLastError(L"GetOverlappedResult");
  }
  CloseHandle(event_handle);
  if (written != data.size()) {
    throw NamedPipeError("partial write detected");
  }
}

std::vector<uint8_t> WindowsPipeHandle::Read() {
  if (handle_ == INVALID_HANDLE_VALUE) {
    throw NamedPipeError("pipe closed");
  }
  std::vector<uint8_t> buffer(4096);
  if (!overlapped_) {
    DWORD bytes_read = 0;
    if (!ReadFile(handle_, buffer.data(), static_cast<DWORD>(buffer.size()),
                  &bytes_read, nullptr)) {
      const DWORD error = GetLastError();
      if (error == ERROR_BROKEN_PIPE || error == ERROR_PIPE_NOT_CONNECTED) {
        return {};
      }
      ThrowLastError(L"ReadFile");
    }
    buffer.resize(bytes_read);
    return buffer;
  }

  HANDLE event_handle = CreateEventW(nullptr, TRUE, FALSE, nullptr);
  if (!event_handle) {
    ThrowLastError(L"CreateEventW");
  }
  OVERLAPPED overlapped{};
  overlapped.hEvent = event_handle;
  DWORD bytes_read = 0;
  BOOL ok = ReadFile(handle_, buffer.data(), static_cast<DWORD>(buffer.size()),
                     nullptr, &overlapped);
  if (!ok) {
    const DWORD error = GetLastError();
    if (error == ERROR_BROKEN_PIPE || error == ERROR_PIPE_NOT_CONNECTED) {
      CloseHandle(event_handle);
      return {};
    }
    if (error != ERROR_IO_PENDING) {
      CloseHandle(event_handle);
      ThrowLastError(L"ReadFile");
    }
    if (WaitForSingleObject(event_handle, INFINITE) != WAIT_OBJECT_0) {
      CloseHandle(event_handle);
      ThrowLastError(L"WaitForSingleObject");
    }
  }
  if (!GetOverlappedResult(handle_, &overlapped, &bytes_read, TRUE)) {
    CloseHandle(event_handle);
    ThrowLastError(L"GetOverlappedResult");
  }
  CloseHandle(event_handle);
  buffer.resize(bytes_read);
  return buffer;
}

WindowsPipeListener::WindowsPipeListener(std::wstring pipe_name,
                                         std::size_t buffer_size)
    : pipe_name_(std::move(pipe_name)), buffer_size_(buffer_size) {}

PipeHandlePtr WindowsPipeListener::Accept() {
  HANDLE pipe = CreateNamedPipeW(
      pipe_name_.c_str(), PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED,
      PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT, PIPE_UNLIMITED_INSTANCES,
      static_cast<DWORD>(buffer_size_), static_cast<DWORD>(buffer_size_), 0,
      nullptr);
  if (pipe == INVALID_HANDLE_VALUE) {
    ThrowLastError(L"CreateNamedPipeW");
  }

  OVERLAPPED overlapped{};
  HANDLE event_handle = CreateEventW(nullptr, TRUE, FALSE, nullptr);
  if (!event_handle) {
    CloseHandle(pipe);
    ThrowLastError(L"CreateEventW");
  }
  overlapped.hEvent = event_handle;

  BOOL connected = ConnectNamedPipe(pipe, &overlapped);
  if (!connected) {
    const DWORD error = GetLastError();
    if (error == ERROR_IO_PENDING) {
      const DWORD wait_result = WaitForSingleObject(event_handle, INFINITE);
      if (wait_result != WAIT_OBJECT_0) {
        CloseHandle(event_handle);
        CloseHandle(pipe);
        ThrowLastError(L"WaitForSingleObject");
      }
      DWORD transferred = 0;
      if (!GetOverlappedResult(pipe, &overlapped, &transferred, FALSE)) {
        CloseHandle(event_handle);
        CloseHandle(pipe);
        ThrowLastError(L"GetOverlappedResult");
      }
    } else if (error == ERROR_PIPE_CONNECTED) {
      SetEvent(event_handle);
    } else {
      CloseHandle(event_handle);
      CloseHandle(pipe);
      ThrowLastError(L"ConnectNamedPipe");
    }
  }

  CloseHandle(event_handle);
  return std::make_unique<WindowsPipeHandle>(pipe, true);
}

std::unique_ptr<PipeListener> CreateNamedPipeListener(
    const NamedPipeListenerOptions& options) {
  if (options.endpoint.empty()) {
    throw NamedPipeError("named pipe endpoint must not be empty");
  }
  std::wstring pipe_name = WideFromUtf8(options.endpoint);
  if (pipe_name.empty()) {
    throw NamedPipeError("failed to convert endpoint to wide string");
  }
  return std::make_unique<WindowsPipeListener>(std::move(pipe_name),
                                               options.buffer_size);
}

PipeHandlePtr ConnectNamedPipeEndpoint(
    const NamedPipeConnectorOptions& options) {
  if (options.endpoint.empty()) {
    throw NamedPipeError("named pipe endpoint must not be empty");
  }
  std::wstring pipe_name = WideFromUtf8(options.endpoint);
  if (pipe_name.empty()) {
    throw NamedPipeError("failed to convert endpoint to wide string");
  }

  auto wait_interval = options.retry_interval;
  if (wait_interval.count() <= 0) {
    wait_interval = std::chrono::milliseconds(50);
  }

  const bool infinite_wait = options.connect_timeout.count() <= 0;
  const auto deadline = infinite_wait
                             ? std::chrono::steady_clock::time_point::max()
                             : std::chrono::steady_clock::now() + options.connect_timeout;

  HANDLE handle = INVALID_HANDLE_VALUE;
  while (true) {
    handle = CreateFileW(pipe_name.c_str(), GENERIC_READ | GENERIC_WRITE, 0, nullptr,
                         OPEN_EXISTING, 0, nullptr);
    if (handle != INVALID_HANDLE_VALUE) {
      break;
    }

    const DWORD error = GetLastError();
    if (error != ERROR_PIPE_BUSY && error != ERROR_FILE_NOT_FOUND) {
      ThrowLastError(L"CreateFileW");
    }

    if (!infinite_wait && std::chrono::steady_clock::now() >= deadline) {
      throw NamedPipeError("connect named pipe timeout");
    }

    using Rep = std::chrono::milliseconds::rep;
    Rep interval_count = wait_interval.count();
    if (interval_count < static_cast<Rep>(1)) {
      interval_count = static_cast<Rep>(1);
    }
    const Rep max_allowed = static_cast<Rep>(std::numeric_limits<DWORD>::max() - 1);
    if (interval_count > max_allowed) {
      interval_count = max_allowed;
    }
    const DWORD wait_ms = static_cast<DWORD>(interval_count);

    WaitNamedPipeW(pipe_name.c_str(), wait_ms);
    std::this_thread::sleep_for(wait_interval);
  }

  DWORD mode = PIPE_READMODE_BYTE;
  if (!SetNamedPipeHandleState(handle, &mode, nullptr, nullptr)) {
    HANDLE to_close = handle;
    CloseHandle(to_close);
    ThrowLastError(L"SetNamedPipeHandleState");
  }

  return std::make_unique<WindowsPipeHandle>(handle, false);
}

void ServeNamedPipe(PipeServer& server, PipeListener& listener,
                    NamedPipeHandshake handshake,
                    std::chrono::milliseconds maintenance_interval,
                    NamedPipeShouldContinue should_continue,
                    std::size_t worker_threads,
                    std::size_t read_buffer_size) {
  if (!should_continue) {
    throw NamedPipeError("should_continue callback must be provided");
  }
  if (!handshake) {
    throw NamedPipeError("handshake callback must be provided");
  }
  if (worker_threads == 0) {
    const auto concurrency = std::thread::hardware_concurrency();
    worker_threads = concurrency == 0 ? 1 : static_cast<std::size_t>(concurrency);
  }

  HANDLE completion_port =
      CreateIoCompletionPort(INVALID_HANDLE_VALUE, nullptr, 0, 0);
  if (!completion_port) {
    ThrowLastError(L"CreateIoCompletionPort");
  }

  std::atomic<bool> running{true};
  std::mutex contexts_mutex;
  std::vector<std::shared_ptr<ConnectionContext>> contexts;
  auto shared_state = std::make_shared<SharedIocpTelemetry>();
  TelemetryCenter& telemetry = server.telemetry();

  auto worker_proc = [&]() {
    while (running.load(std::memory_order_relaxed)) {
      DWORD bytes_transferred = 0;
      ULONG_PTR completion_key = 0;
      OVERLAPPED* overlapped = nullptr;
      const BOOL ok = GetQueuedCompletionStatus(completion_port, &bytes_transferred,
                                                &completion_key, &overlapped,
                                                INFINITE);
      if (!ok && overlapped == nullptr) {
        if (!running.load(std::memory_order_relaxed)) {
          break;
        }
        telemetry.IncrementCounter("ipc_winpipe_completion_port_error_total");
        server.ReportError("GetQueuedCompletionStatus failed");
        continue;
      }
      if (overlapped == nullptr) {
        break;
      }
      auto* op = reinterpret_cast<IocpOperation*>(overlapped);
      auto ctx = op->context;
      if (!ok) {
        const DWORD error = GetLastError();
        ctx->active_operations.fetch_sub(1, std::memory_order_relaxed);
        ctx->UpdatePendingOpsGauge();
        delete op;
        if (error != ERROR_OPERATION_ABORTED) {
          ReportIocpError(ctx, "I/O error");
          CloseConnection(ctx, "I/O error");
        }
        FinalizeIfIdle(ctx);
        continue;
      }
      if (op->type == OperationType::kRead) {
        HandleReadCompletion(op, bytes_transferred);
      } else {
        HandleWriteCompletion(op);
      }
    }
  };

  std::vector<std::thread> workers;
  workers.reserve(worker_threads);
  for (std::size_t i = 0; i < worker_threads; ++i) {
    workers.emplace_back(worker_proc);
  }

  auto cleanup_contexts = [&]() {
    std::lock_guard<std::mutex> lock(contexts_mutex);
    contexts.erase(std::remove_if(contexts.begin(), contexts.end(),
                                  [](const std::shared_ptr<ConnectionContext>& ctx) {
                                    return ctx->closing.load(std::memory_order_relaxed) &&
                                           ctx->active_operations.load(std::memory_order_relaxed) == 0 &&
                                           !ctx->maintenance_thread.joinable();
                                  }),
                    contexts.end());
  };

  while (should_continue()) {
    cleanup_contexts();
    PipeHandlePtr base_handle;
    try {
      base_handle = listener.Accept();
      telemetry.IncrementCounter("ipc_winpipe_accept_total");
    } catch (const std::exception& ex) {
      telemetry.IncrementCounter("ipc_winpipe_accept_error_total");
      server.ReportError(ex.what());
      cleanup_contexts();
      continue;
    }

    SessionContext session;
    bool handshake_ok = false;
    std::string handshake_error;
    try {
      handshake_ok = handshake(server, *base_handle, session);
    } catch (const std::exception& ex) {
      handshake_error = ex.what();
    } catch (...) {
      handshake_error = "unknown handshake exception";
    }

    if (!handshake_ok) {
      const bool shutting_down = !should_continue();
      if (!shutting_down) {
        telemetry.IncrementCounter("ipc_winpipe_handshake_failure_total");
        if (!handshake_error.empty()) {
          server.ReportError("handshake failed: " + handshake_error);
        } else {
          server.ReportError("handshake rejected by callback");
        }
      }
      continue;
    }

    telemetry.IncrementCounter("ipc_winpipe_handshake_success_total");

    auto raw_handle = base_handle.release();
    auto* win_handle = dynamic_cast<WindowsPipeHandle*>(raw_handle);
    if (!win_handle) {
      delete raw_handle;
      throw NamedPipeError("expected WindowsPipeHandle after accept");
    }
    std::unique_ptr<WindowsPipeHandle> handle_ptr(win_handle);

    if (!CreateIoCompletionPort(handle_ptr->native_handle(), completion_port, 0,
                                0)) {
      ThrowLastError(L"CreateIoCompletionPort");
    }

    auto channel = server.CreateChannel(std::move(session));

    auto ctx = std::make_shared<ConnectionContext>(
        server, std::move(handle_ptr), std::move(channel), completion_port,
        maintenance_interval, read_buffer_size, shared_state);

    ctx->ReportConnectionOpen();
    ctx->StartMaintenance();
    if (!SubmitRead(ctx)) {
      ctx->StopMaintenance();
      continue;
    }

    {
      std::lock_guard<std::mutex> lock(contexts_mutex);
      contexts.push_back(ctx);
    }
  }

  running.store(false, std::memory_order_relaxed);
  {
    std::lock_guard<std::mutex> lock(contexts_mutex);
    for (auto& ctx : contexts) {
      CloseConnection(ctx, "shutdown");
    }
  }

  for (std::size_t i = 0; i < worker_threads; ++i) {
    PostQueuedCompletionStatus(completion_port, 0, 0, nullptr);
  }

  for (auto& worker : workers) {
    if (worker.joinable()) {
      worker.join();
    }
  }

  {
    std::lock_guard<std::mutex> lock(contexts_mutex);
    for (auto& ctx : contexts) {
      ctx->StopMaintenance();
    }
    contexts.clear();
  }

  CloseHandle(completion_port);
}

}  // namespace asycipc

#elif !defined(__linux__) && !defined(__APPLE__)

#include <stdexcept>

namespace asycipc {

std::unique_ptr<PipeListener> CreateNamedPipeListener(
    const NamedPipeListenerOptions& options) {
  (void)options;
  throw NamedPipeError("named pipes are not supported on this platform");
}

PipeHandlePtr ConnectNamedPipeEndpoint(
    const NamedPipeConnectorOptions& options) {
  (void)options;
  throw NamedPipeError("named pipes are not supported on this platform");
}

void ServeNamedPipe(PipeServer&, PipeListener&, NamedPipeHandshake,
                    std::chrono::milliseconds, NamedPipeShouldContinue,
                    std::size_t, std::size_t) {
  throw NamedPipeError("named pipes are not supported on this platform");
}

}  // namespace asycipc

#endif  // _WIN32

